package com.xxx.rabbitmq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * 【可以不用提前声明，在消费端声明时会自动创建交换机和队列】
 */
@Slf4j
@Configuration
//public class RabbitMqConfig implements ApplicationContextAware {  // 修改rabbitTemplate方式2：使用ApplicationContextAware
public class RabbitMqConfig {

    // 修改rabbitTemplate方式1：
    @Bean
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        // 当需要确保消息被可靠投递时，例如：金融交易、订单处理，可以通过以下方式确保
        // 1.设置 Confirm 回调（用于确认消息是否到达交换机）
        // 需要在application.yml里配置：
        //    publisher-confirms: true  # 开启消息异步confirm模式
        log.info("配置publisher 异步confirm模式（全局单例）");
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {

            @Override
            public void confirm(CorrelationData cd, boolean ack, String cause) {
                log.debug("收到消息发送异步确认， 消息ID:{}, ack:{}, cause:{}", cd.getId(), ack, cause);
            }
        });

        // 2.设置 Return 回调（用于确认消息是否到达交换机并且被路由到队列，当消息无法路由到队列时触发）
        // 需要在application.yml里配置：
        //    publisher-confirm-type: correlated   # 开启消息发送确认机制，并设置confirm类型
        //    publisher-returns: true   # 开启消息发送 return机制
        log.info("配置publisher return回调（全局单例）");
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyMsg, String exchange, String key) {
                log.debug("收到消息发送回调， replyCode:{}, replyMsg:{}, exchange:{}, routingKey:{}", replyCode, replyMsg, exchange, key);
            }
        });

//        // 3.启用事务模式（和消息发送确认模式不可同时启用）
//        log.debug("启用事务模式");
//        rabbitTemplate.setChannelTransacted(true);

        return rabbitTemplate;
    }

//    // 开启事务时，必须要配置事务管理器
//    @Bean
//    public PlatformTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {
//        return new RabbitTransactionManager(connectionFactory);
//    }

    // 修改rabbitTemplate方式2
//    /**
//     * 将修改属性后的rabbitTemplate Bean注入到spring的上下文applicationContext中(ApplicationContextAware在初始化Spring容器后执行)
//     * @param applicationContext
//     * @throws BeansException
//     */
//    @Override
//    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
//        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
//
//        // 当需要确保消息被可靠投递时，例如：金融交易、订单处理，可以通过以下方式确保
//        // 1.设置 Confirm 回调（用于确认消息是否到达交换机）
//        // 需要在application.yml里配置：
//        //    publisher-confirms: true  # 开启消息异步confirm模式
//        log.info("配置publisher 异步confirm模式（全局单例）");
//        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
//
//            @Override
//            public void confirm(CorrelationData cd, boolean ack, String cause) {
//                log.debug("收到消息发送异步确认， 消息ID:{}, ack:{}, cause:{}", cd.getId(), ack, cause);
//            }
//        });
//
//        // 2.设置 Return 回调（用于确认消息是否到达交换机并且被路由到队列，当消息无法路由到队列时触发）
//        // 需要在application.yml里配置：
//        //    publisher-confirm-type: correlated   # 开启消息发送确认机制，并设置confirm类型
//        //    publisher-returns: true   # 开启消息发送 return机制
//        log.info("配置publisher return回调（全局单例）");
//        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
//            @Override
//            public void returnedMessage(Message message, int replyCode, String replyMsg, String exchange, String key) {
//                log.debug("收到消息发送回调， replyCode:{}, replyMsg:{}, exchange:{}, routingKey:{}", replyCode, replyMsg, exchange, key);
//            }
//        });
//    }

    /**
     * JSON消息转换器（针对发送对象类型的消息）
     */
    @Bean
    public MessageConverter jsonMessageConverter(){
        return new Jackson2JsonMessageConverter();
    }

    /**
     * 普通消息转换器
     */
    @Bean
    public MessageConverter messageConverter(){
        Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
        jjmc.setCreateMessageIds(true);   // 给消息加上ID
        return jjmc;
    }

    ///////////////////////////////// direct消息 start ///////////////////////////////
    /**
     * 声明direct交换机
     */
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange("direct.exchange");
    }

    /**
     * 声明direct队列：direct.queue1
     */
    @Bean
    public Queue directQueue1(){
        return new Queue("direct.queue1");
    }

    /**
     * 将direct.queue1队列绑定到交换机上
     */
    @Bean
    public Binding bindingDirectQueue1(Queue directQueue1, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue1).to(directExchange).with("routingK");
    }

    /**
     * 声明direct另一个队列：direct.queue2
     */
    @Bean
    public Queue directQueue2(){
        return new Queue("direct.queue2");
    }

    /**
     * 将direct另一个队列绑定到direct交换机上
     */
    @Bean
    public Binding bindingDirectQueue2(Queue directQueue2, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue2).to(directExchange).with("routingK");
    }
    ///////////////////////////////// direct消息 end ///////////////////////////////


    ///////////////////////////////// fanout消息 start ///////////////////////////////
    /**
     * 声明fanout交换机
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanout.exchange");
    }

    /**
     * 声明fanout队列：fanout.queue1
     */
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

    /**
     * 将fanout.queue1队列绑定到交换机上
     */
    @Bean
    public Binding bindingFanoutQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    /**
     * 声明fanout另一个队列：fanout.queue2
     */
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

    /**
     * 将fanout另一个队列绑定到fanout交换机上
     */
    @Bean
    public Binding bindingFanoutQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
    ///////////////////////////////// fanout消息 end ///////////////////////////////


    ///////////////////////////////// topic消息 start ///////////////////////////////
    /**
     * 声明topic交换机
     */
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange("topic.exchange");
    }

    /**
     * 声明topic队列：topic.queue1
     */
    @Bean
    public Queue topicQueue1(){
        return new Queue("topic.queue1");
    }

    /**
     * topic.queue1队列绑定到交换机上
     */
    @Bean
    public Binding bindingTopicQueue1(Queue topicQueue1, TopicExchange topicExchange) {
        return BindingBuilder.bind(topicQueue1).to(topicExchange).with("china.#");
    }

    /**
     * 声明topic另一个队列：topic.queue2
     */
    @Bean
    public Queue topicQueue2(){
        return new Queue("topic.queue2");
    }

    /**
     * 将另一个队列绑定到direct交换机上
     */
    @Bean
    public Binding bindingTopicQueue2(Queue topicQueue2, TopicExchange topicExchange) {
        return BindingBuilder.bind(topicQueue2).to(topicExchange).with("red");
    }
    ///////////////////////////////// topic消息 end ///////////////////////////////


    ///////////////////////////////// 过期消息 start ///////////////////////////////
//    /**
//     * 声明过期消息队列交换机（如果在消费者端已经声明过，这里就不用再创建了）
//     */
//    @Bean
//    public DirectExchange ttlExchange() {
//        return new DirectExchange("ttl.exchange");
//    }
//
//    /**
//     * 声明过期消息队列
//     */
//    @Bean
//    public Queue ttlQueue(){
//        return QueueBuilder.durable("ttl.queue")        // 指定过期队列名称，并持久化
//                .ttl(10000)                             // 过期时间：10s
//                .deadLetterExchange("dl.exchange")      // 将延时队列绑定死信交换机
//                .deadLetterRoutingKey("dl")             // 延时队列绑定死信交换机的路由key
//                .build();
//    }
//
//    /**
//     * 将过期消息队列绑定到交换机
//     */
//    @Bean
//    public Binding ttlBinding(DirectExchange ttlExchange, Queue ttlQueue){
//        return BindingBuilder.bind(ttlQueue).to(ttlExchange).with("ttl");
//    }
    ///////////////////////////////// 过期消息 end ///////////////////////////////

    ///////////////////////////////// 基于插件的延时队列 start ///////////////////////////////
    /**
     * 声明延时交换机
     */
    @Bean
    CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange("delay.exchange", "x-delayed-message", true, false, args);
    }

    /**
     * 声明延时队列
     */
    @Bean
    public Queue delayQueue() {
        return new Queue("delay.queue", true, false, false);
    }

    /**
     * 将延时队列绑定到延时交换机
     */
    @Bean
    public Binding binding(CustomExchange delayExchange, Queue delayQueue) {
        return BindingBuilder.bind(delayQueue)
                .to(delayExchange)
                .with("delay")
                .noargs();

    }
    ///////////////////////////////// 基于插件的延时队列 end ///////////////////////////////

}
